package com.sskj.kafka.v1;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

/**
 * @author MaSiyi
 * @version 1.0.0 2021/12/10
 * @since JDK 1.8.0
 */
@Service
@Slf4j
public class KafkaConsumer {

    /** 消费监听
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"topictest1"},groupId = "consumer-group")
    public void message1(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("点对点消费1："+record.topic()+"-"+record.partition()+"-"+record.value());
    }

    /** 发布订阅模式
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"topictest2"},groupId = "1")
    public void message2(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("发布订阅模式1："+record.topic()+"-"+record.partition()+"-     "+record.value());
    }
    /** 发布订阅模式
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"topictest2"},groupId = "2")
    public void message3(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("发布订阅模式2："+record.topic()+"-"+record.partition()+"-"+record.value());
    }

    /** 发布订阅模式
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"vp_ba_jam"},groupId = "3")
    public void message4(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("发布订阅模式2："+record.topic()+"-"+record.partition()+"-                         --"+record.value());
    }

    /** 发布订阅模式
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"vp_ba_crossline"},groupId = "3")
    public void message5(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("发布订阅模式2："+record.topic()+"-"+record.partition()+"-                         --"+record.value());
    }

    /** 发布订阅模式
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"vp_ba_stop"},groupId = "3")
    public void message6(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("发布订阅模式2："+record.topic()+"-"+record.partition()+"-                         --"+record.value());
    }

    /** 发布订阅模式
     * @Param: [record]
     * @return: void
     * @Author: MaSiyi
     * @Date: 2021/12/10
     */
//    @KafkaListener(topics = {"vp_ba_result"},groupId = "3")
    public void message7(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("发布订阅模式2："+record.topic()+"-"+record.partition()+"-                         --"+record.value());
    }


    /**
     * 算法识别数据的结果
     * 业务：根据不同类型的算法结果，去做相应的业务
     */
    @KafkaListener(topics = {"vp_ba_result","vp_ba_jam","vp_ba_stop","vp_ba_crossline","vp_tracking"}, groupId = "1111111")
    public void subAlgorithmResult4(ConsumerRecord<?, ?> record) {
        String message = (String) record.value();
        log.info("topic:{} payload:{}", record.topic(), message);
//        AlgorithmResultDto algorithmResultDto = null;
//        try {
//            algorithmResultDto = JsonUtils.parseObject(message, AlgorithmResultDto.class);
//        } catch (Exception e) {
//            log.error("算法识别数据的结果", e);
//            throw new RuntimeException(e);
//        }
//        accessAlgorithmService.processAlgorithmResult(algorithmResultDto);
    }

    /**
     * 推送算法配置
     */
    @KafkaListener(topics = {"cameralist"}, groupId = "222222")
    public void subAlgorithmResult5(ConsumerRecord<?, ?> record) {
        String message = (String) record.value();
        log.info("topic:{} payload:{}", record.topic(), message);
//        AlgorithmResultDto algorithmResultDto = null;
//        try {
//            algorithmResultDto = JsonUtils.parseObject(message, AlgorithmResultDto.class);
//        } catch (Exception e) {
//            log.error("算法识别数据的结果", e);
//            throw new RuntimeException(e);
//        }
//        accessAlgorithmService.processAlgorithmResult(algorithmResultDto);
    }


}

